package com.junjie.blibli.util;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.CountDownLatch2;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

public class RocketMQUtil {
    //同步发送消息，关心消息有没有成功发送出去用这个
    public static void syncSendMsg(DefaultMQProducer producer,Message message) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        SendResult result = producer.send(message);
        System.out.println(result);
    }

    //异步发送消息，不关心回调回值可以用这个
    public static void asyncSendMsg(DefaultMQProducer producer,Message message) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        //这里我们可以做复杂点，做个发送两次信息
        int messageCount = 2;
        //倒计时类器
        CountDownLatch2 countDownLatch2 = new CountDownLatch2(messageCount);
        for(int i=0;i<messageCount;i++){
            //回调函数，发送成功的回调，发送失败的提醒
            producer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    //如果成功了，计数器就要减一次
                    countDownLatch2.countDown();
                    System.out.println(sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable e) {
                    //如果失败了，计数器也可以减一次
                    countDownLatch2.countDown();
                    System.out.println("发送消息的时候发生了异常！"+e);
                }
            });
        }
        //这里可以让定时器停留五秒钟
        countDownLatch2.await(5, TimeUnit.SECONDS);
    }

}
